package com.course.controller;

import com.alibaba.fastjson.JSONObject;
//import com.course.configs.KafkaProducer;
import com.course.configs.RabbitProducer;
import com.course.model.CourseOrder;
import com.course.model.CourseVieData;
import com.course.result.ResultCode;
import com.course.result.ResultData;
import com.course.service.CourseOrderService;
import com.course.service.CourseVieDataService;
import com.course.service.RedisService;
import com.course.utils.CommonUtil;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.servlet.http.HttpServletRequest;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Controller
@RequestMapping("viecourse")
public class VieCourseController implements InitializingBean {

    /**
     * 单机时，使用内存操作判定课程存留
     */
    private static Map<Integer, Boolean> course_is_over = new ConcurrentHashMap<>();

    /**
     * 令牌桶限流
     */
    RateLimiter rateLimiter = RateLimiter.create(10);
    @Autowired
    RedisService redisService;
    @Autowired
    CourseOrderService courseOrderService;
    @Autowired
    CourseVieDataService courseVieDataService;
    //@Autowired
    //KafkaProducer kafkaProducer;
    @Autowired
    RabbitProducer rabbitProducer;

    @RequestMapping(value = "/doVie", method = RequestMethod.POST)
    @ResponseBody
    public ResultData<Integer> vieCourses(HttpServletRequest request) throws Exception{
        if(rateLimiter.tryAcquire(1000)){
            return ResultData.error(ResultCode.ACCESS_LIMIT_REACHED);
        }
            Integer c_id = CommonUtil.getParam(request,"courseId",0);
        Integer userId = 1;
        if(c_id <= 0 || !course_is_over.containsKey(c_id)){
            return ResultData.error(ResultCode.BIND_ERROR);
        }
        if(course_is_over.get(c_id)){
            return ResultData.error(ResultCode.ACCESS_LIMIT_REACHED);
        }
        // 逻辑预减少
        Long now_desc = redisService.decr("Course_"+c_id,1L);
        if(now_desc < 0){
            course_is_over.put(c_id,true);
            return ResultData.error(ResultCode.COURSE_OVER);
        }
        CourseOrder co = courseOrderService.findByUserCourse(userId,c_id);
        if(co != null){
            return ResultData.error(ResultCode.REPEATE_SECKILL);
        }
        // 构建消息
        JSONObject json = new JSONObject();
        json.put("userId",userId);
        json.put("courseId",c_id);
        // 放入消息队列
        //kafkaProducer.sendChannelMess("test",json.toString());
        rabbitProducer.sendMessage("test",json.toString());
        // 等待
        return ResultData.success(0);
    }
  /*  @RequestMapping(value = "/test")
    public void kafkaTest(HttpServletRequest request) throws Exception{
        // 放入消息队列
        kafkaProducer.sendChannelMess("test","test");
        // 等待
    }*/

    /**
     * 初始化数据
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        List<CourseVieData> list = courseVieDataService.findVieCourseData(System.currentTimeMillis());
        for (CourseVieData cvd : list){
            redisService.setVal("Course_"+cvd.getCourseId(),cvd.getApplyNums());
            course_is_over.put(cvd.getCourseId(),false);
        }
    }
}
